package com.example.springboot3.disruptor;

import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;

import java.nio.ByteBuffer;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

public class LongEventProducer {
    private final RingBuffer<LongEvent> ringBuffer;

    public LongEventProducer(RingBuffer<LongEvent> ringBuffer) {
        this.ringBuffer = ringBuffer;
    }

    public void onData(long value) {
        long sequence = ringBuffer.next();
        try {
            LongEvent event = ringBuffer.get(sequence);
            event.setValue(value);
        } finally {
            ringBuffer.publish(sequence);
        }
    }
}

 class LongEventConsumerHandler implements EventHandler<LongEvent> {
    public void onEvent(LongEvent event, long sequence, boolean endOfBatch) {
        System.out.println("事件值: " + event.getValue());
    }


     public static void main(String[] args) throws InterruptedException {
         Executor executor = Executors.newCachedThreadPool();
         LongEventFactory factory = new LongEventFactory();
         int bufferSize = 1024;

         Disruptor<LongEvent> disruptor = new Disruptor<>(factory, bufferSize, (ThreadFactory) executor);
         disruptor.handleEventsWith(new LongEventConsumerHandler());
         disruptor.start();

         RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
         LongEventProducer producer = new LongEventProducer(ringBuffer);

         ByteBuffer bb = ByteBuffer.allocate(8);
         for (long l = 0; l<100; l++) {
             bb.putLong(0, l);
             producer.onData(bb.getLong(0));
             // 延迟一秒
             Thread.sleep(1000);
         }
     }
 }

